package com.dosgi.kit;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dosgi.conf.DosgiPoolConf;


/**
 * @author dingnate
 *
 */
public class ThreadPoolKit {
	private static transient final Logger LOG = LoggerFactory.getLogger(ThreadPoolKit.class);
	static final ThreadPoolExecutor exec;
	static {
		// 初始化普通线程池
		int capacity = ValueKit.getValue(DosgiPoolConf.me.getWorkQueueCapacity(), 100);
		int maximumPoolSize = ValueKit.getValue(DosgiPoolConf.me.getMaximumPoolSize(), Runtime.getRuntime()
				.availableProcessors() * 2);
		long keepAliveTime = ValueKit.getValue(DosgiPoolConf.me.getKeepAliveTime(), 3L);
		TimeUnit timeUnit = TimeUnit.valueOf(ValueKit.getValue(DosgiPoolConf.me.getTimeUnit(), TimeUnit.MINUTES.name()));
		String prefix = ValueKit.getValue(DosgiPoolConf.me.getPrefix(), "");
		exec = newThreadPool(capacity, maximumPoolSize, keepAliveTime, timeUnit, prefix);
	}

	/**
	 * 创建自动收缩的线程池
	 * 
	 * @param workQueueCapacity
	 *            工作队列超过workQueueCapacity时，新提交的任务会被拒绝
	 * @param maximumPoolSize
	 *            最大线程数
	 * @param keepAliveTime
	 *            空闲线程生存时间
	 * @param timeUnit
	 *            时间单位{@link TimeUnit}的枚举值的名字
	 * @param prefix
	 *            线程前缀
	 * @return
	 */
	public static ThreadPoolExecutor newThreadPool(int workQueueCapacity, int maximumPoolSize, long keepAliveTime,
			TimeUnit timeUnit, final String prefix) {
		ThreadPoolExecutor executor = new ThreadPoolExecutor(maximumPoolSize, maximumPoolSize, keepAliveTime, timeUnit,
				new ArrayBlockingQueue<Runnable>(workQueueCapacity > 0 ? workQueueCapacity : 100),
				StrKit.isBlank(prefix) ? Executors.defaultThreadFactory() : new ThreadFactory() {
					private final AtomicInteger threadNumber = new AtomicInteger(0);

					@Override
					public Thread newThread(Runnable r) {
						return new Thread(r, prefix + threadNumber.incrementAndGet());
					}
				}, new RejectedExecutionHandler() {
					@Override
					public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
						String msg = String.format(
								"ThreadPoolKit Task %s  rejected by %s\nCaused by workQueue is full.", r.toString(),
								executor.toString());
						RejectedExecutionException t = new RejectedExecutionException(msg);
						LOG.error(msg);
						throw t;
					}
				});
		executor.allowCoreThreadTimeOut(true);
		return executor;
	}

	ThreadPoolKit() {
	}

	/**
	 * Submits a value-returning task for execution and returns a Future
	 * representing the pending results of the task. The Future's {@code get}
	 * method will return the task's result upon successful completion.
	 *
	 * <p>
	 * If you would like to immediately block waiting for a task, you can use
	 * constructions of the form {@code result = exec.submit(aCallable).get();}
	 *
	 * <p>
	 * Note: The {@link Executors} class includes a set of methods that can
	 * convert some other common closure-like objects, for example,
	 * {@link java.security.PrivilegedAction} to {@link Callable} form so they
	 * can be submitted.
	 *
	 * @param task
	 *            the task to submit
	 * @param <T>
	 *            the type of the task's result
	 * @return a Future representing pending completion of the task
	 * @throws RejectedExecutionException
	 *             if the task cannot be scheduled for execution
	 * @throws NullPointerException
	 *             if the task is null
	 */
	public static <V> Future<V> submit(Callable<V> call) {
		return exec.submit(call);
	}

	/**
	 * Submits a Runnable task for execution and returns a Future representing
	 * that task. The Future's {@code get} method will return {@code null} upon
	 * <em>successful</em> completion.
	 *
	 * @param task
	 *            the task to submit
	 * @return a Future representing pending completion of the task
	 * @throws RejectedExecutionException
	 *             if the task cannot be scheduled for execution
	 * @throws NullPointerException
	 *             if the task is null
	 */
	public static Future<?> submit(Runnable task) {
		return exec.submit(task);
	}

	/**
	 * Executes the given tasks, returning a list of Futures holding their
	 * status and results when all complete. {@link Future#isDone} is
	 * {@code true} for each element of the returned list. Note that a
	 * <em>completed</em> task could have terminated either normally or by
	 * throwing an exception. The results of this method are undefined if the
	 * given collection is modified while this operation is in progress.
	 *
	 * @param tasks
	 *            the collection of tasks
	 * @param <T>
	 *            the type of the values returned from the tasks
	 * @return a list of Futures representing the tasks, in the same sequential
	 *         order as produced by the iterator for the given task list, each
	 *         of which has completed
	 * @throws InterruptedException
	 *             if interrupted while waiting, in which case unfinished tasks
	 *             are cancelled
	 * @throws NullPointerException
	 *             if tasks or any of its elements are {@code null}
	 * @throws RejectedExecutionException
	 *             if any task cannot be scheduled for execution
	 */
	public static List<Future<Object>> invokeAll(Collection<? extends Callable<Object>> tasks) throws Exception {
		return exec.invokeAll(tasks);
	}

	/**
	 * Executes the given tasks, returning a list of Futures holding their
	 * status and results when all complete or the timeout expires, whichever
	 * happens first. {@link Future#isDone} is {@code true} for each element of
	 * the returned list. Upon return, tasks that have not completed are
	 * cancelled. Note that a <em>completed</em> task could have terminated
	 * either normally or by throwing an exception. The results of this method
	 * are undefined if the given collection is modified while this operation is
	 * in progress.
	 *
	 * @param tasks
	 *            the collection of tasks
	 * @param timeout
	 *            the maximum time to wait
	 * @param unit
	 *            the time unit of the timeout argument
	 * @param <T>
	 *            the type of the values returned from the tasks
	 * @return a list of Futures representing the tasks, in the same sequential
	 *         order as produced by the iterator for the given task list. If the
	 *         operation did not time out, each task will have completed. If it
	 *         did time out, some of these tasks will not have completed.
	 * @throws InterruptedException
	 *             if interrupted while waiting, in which case unfinished tasks
	 *             are cancelled
	 * @throws NullPointerException
	 *             if tasks, any of its elements, or unit are {@code null}
	 * @throws RejectedExecutionException
	 *             if any task cannot be scheduled for execution
	 */
	public static List<Future<Object>> invokeAll(Collection<? extends Callable<Object>> tasks, long timeout,
			TimeUnit unit) throws Exception {
		return exec.invokeAll(tasks, timeout, unit);
	}

	/**
	 * Executes the given tasks, returning the result of one that has completed
	 * successfully (i.e., without throwing an exception), if any do. Upon
	 * normal or exceptional return, tasks that have not completed are
	 * cancelled. The results of this method are undefined if the given
	 * collection is modified while this operation is in progress.
	 *
	 * @param tasks
	 *            the collection of tasks
	 * @param <T>
	 *            the type of the values returned from the tasks
	 * @return the result returned by one of the tasks
	 * @throws InterruptedException
	 *             if interrupted while waiting
	 * @throws NullPointerException
	 *             if tasks or any element task subject to execution is
	 *             {@code null}
	 * @throws IllegalArgumentException
	 *             if tasks is empty
	 * @throws ExecutionException
	 *             if no task successfully completes
	 * @throws RejectedExecutionException
	 *             if tasks cannot be scheduled for execution
	 */
	public static Object invokeAny(Collection<? extends Callable<Object>> tasks) throws Exception {
		return exec.invokeAny(tasks);
	}

	/**
	 * Executes the given tasks, returning the result of one that has completed
	 * successfully (i.e., without throwing an exception), if any do before the
	 * given timeout elapses. Upon normal or exceptional return, tasks that have
	 * not completed are cancelled. The results of this method are undefined if
	 * the given collection is modified while this operation is in progress.
	 *
	 * @param tasks
	 *            the collection of tasks
	 * @param timeout
	 *            the maximum time to wait
	 * @param unit
	 *            the time unit of the timeout argument
	 * @param <T>
	 *            the type of the values returned from the tasks
	 * @return the result returned by one of the tasks
	 * @throws InterruptedException
	 *             if interrupted while waiting
	 * @throws NullPointerException
	 *             if tasks, or unit, or any element task subject to execution
	 *             is {@code null}
	 * @throws TimeoutException
	 *             if the given timeout elapses before any task successfully
	 *             completes
	 * @throws ExecutionException
	 *             if no task successfully completes
	 * @throws RejectedExecutionException
	 *             if tasks cannot be scheduled for execution
	 */
	public static Object invokeAny(Collection<? extends Callable<Object>> tasks, long timeout, TimeUnit unit)
			throws Exception {
		return exec.invokeAny(tasks, timeout, unit);
	}

	public static void stop() {
		exec.shutdownNow();
	}
}
